package com.zhan.signalling.store.tablestore;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alicloud.openservices.tablestore.ClientConfiguration;
import com.alicloud.openservices.tablestore.model.AlwaysRetryStrategy;
import com.alicloud.openservices.tablestore.model.CreateTableRequest;
import com.alicloud.openservices.tablestore.model.DefinedColumnSchema;
import com.alicloud.openservices.tablestore.model.IndexMeta;
import com.alicloud.openservices.tablestore.model.PrimaryKeySchema;
import com.alicloud.openservices.tablestore.model.TableMeta;
import com.alicloud.openservices.tablestore.model.TableOptions;
import com.alicloud.openservices.tablestore.timeline.message.IMessage;
import com.alicloud.openservices.tablestore.timeline.store.DistributeTimelineConfig;
import com.alicloud.openservices.tablestore.timeline.store.DistributeTimelineStore;
import com.alicloud.openservices.tablestore.timeline.store.IStore;
import com.zhan.signalling.entity.ChatRecord;
import com.zhan.signalling.entity.TableStoreBeanDefinition;
import com.zhan.signalling.entity.TableStoreBeanDefinition.Column;
import com.zhan.signalling.entity.WhiteBoardRecord;
import com.zhan.signalling.server.SocketServer;
import com.zhan.signalling.store.StoreMessageStore;
import com.zhan.signalling.store.tablestore.message.ChatMessage;
import com.zhan.signalling.store.tablestore.message.WhiteBoardMessage;
import com.zhan.signalling.util.BeanUtil;

public class TableStoreMessageStore extends StoreMessageStore {
	
	private final static Logger LOG = LoggerFactory.getLogger(TableStoreMessageStore.class);
	
	private String endPoint;
	private String accessKeyId;
	private String accessKeySecret;
	private String instanceName;
	
	public final static String name = "tablestore";
	
	private IStore store;
	
	private MessageOperation messageOperation;
	
	private TableStoreMessageStore() {
	}

	public static StoreMessageStore getHandler(Properties prop) {
		
		String endPoint = prop.getProperty(SocketServer.STORE_MESSAGE_PROP+".endPoint");
		final String accessKeyId = prop.getProperty(SocketServer.STORE_MESSAGE_PROP+".accessKeyId");
		final String accessKeySecret = prop.getProperty(SocketServer.STORE_MESSAGE_PROP+".accessKeySecret");
		final String instanceName = prop.getProperty(SocketServer.STORE_MESSAGE_PROP+".instanceName");
		TableStoreMessageStore handler = new TableStoreMessageStore();
		handler.endPoint = endPoint;
		handler.accessKeyId = accessKeyId;
		handler.accessKeySecret = accessKeySecret;
		handler.instanceName = instanceName;
		return handler;
	};
	
	public void initStore(Class<?> tableClass) {
		TableStoreBeanDefinition beanDefinition = TableStoreBeanDefinition.build(tableClass);
		List<Column> columns = beanDefinition.getColumns();
		// 获取主键 
		Column firstPK = columns.get(0);
		
		ClientConfiguration clientConfiguration = new ClientConfiguration();
        // 设置建立连接的超时时间。
        clientConfiguration.setConnectionTimeoutInMillisecond(5000);
        // 设置socket超时时间。
        clientConfiguration.setSocketTimeoutInMillisecond(5000);
        // 设置重试策略，若不设置，采用默认的重试策略。
        clientConfiguration.setRetryStrategy(new AlwaysRetryStrategy());
        
        // 消息存储
        DistributeTimelineConfig storeConfig = new DistributeTimelineConfig(endPoint, accessKeyId, accessKeySecret, instanceName, 
        		beanDefinition.getTableName());
        storeConfig.setFirstPKName(firstPK.getName());
        storeConfig.setTtl(-1);
        storeConfig.setMessageContentSuffix("msg");
        storeConfig.setClientConfiguration(clientConfiguration);
        
        store = new DistributeTimelineStore(storeConfig);
        messageOperation = new MessageOperation(store);
        //消息同步
//        DistributeTimelineConfig syncConfig = new DistributeTimelineConfig(endPoint, accessKeyId, accessKeySecret, instanceName, 
//        		"temp_" + beanDefinition.getTableName());	
//		syncConfig.setFirstPKName(firstPK.getName());
//		syncConfig.setLimit(1000);
//		syncConfig.setTtl(24 * 60 * 60);
//		syncConfig.setMessageContentSuffix("msg");
//		syncConfig.setClientConfiguration(clientConfiguration);
//        sync = new DistributeTimelineStore(syncConfig);
		
        
        
	}
	
	@Deprecated
	private void createTable(Class<?> tableClass) {
		TableStoreBeanDefinition beanDefinition = TableStoreBeanDefinition.build(tableClass);
		// 判断表是否存在
		if(existTable(beanDefinition.getTableName())) {
			LOG.warn("table[{}] is exist",beanDefinition.getTableName());
			return;
		}
		LOG.info("create table[{}] start...",beanDefinition.getTableName());	
		// 定义表元数据
        final TableMeta tableMeta = new TableMeta(beanDefinition.getTableName());
        
        final List<IndexMeta> indexMetas = new ArrayList<>();
        List<Column> columnList =  beanDefinition.getColumns();
        columnList.stream().forEach(column -> {
        	String name = column.getName();
        	Class<?> type = column.getClazz();
        	boolean isPrimaryKey = column.isPrimaryKey();
        	boolean isIndexColumn = column.isIndexColumn();
        	if(isPrimaryKey) {
        		tableMeta.addPrimaryKeyColumn(new PrimaryKeySchema(name, column.getPKType(type)));
        	}else if(isIndexColumn) {
        		String index_name = column.getIndexName();
        		final List<String> pkNames = column.getAssociationPrimaryKey();
        		IndexMeta indexMeta = new IndexMeta(index_name);
        		pkNames.forEach(pkName -> indexMeta.addPrimaryKeyColumn(pkName)); // 为索引表添加主键列。
        		indexMeta.addDefinedColumn(name); // 为索引表添加属性列。
        	}else {
        		tableMeta.addDefinedColumn(new DefinedColumnSchema(name, column.getDefinedColumnType(type)));
        	}
        });
        
        int timeToLive = beanDefinition.getTimeToLive(); // 数据的过期时间, 单位秒, -1代表永不过期. 带索引表的主表数据过期时间必须为-1。
        int maxVersions = beanDefinition.getMaxVersion(); // 保存的最大版本数, 带索引表的请表最大版本数必须为1。
        TableOptions tableOptions = new TableOptions(timeToLive, maxVersions);
       
        // 创建主表，索引表
        CreateTableRequest request = null;
        if(CollectionUtils.isEmpty(indexMetas)) {
        	request = new CreateTableRequest(tableMeta, tableOptions); // 创建主表。
        }else {
        	request = new CreateTableRequest(tableMeta, tableOptions, indexMetas); // 创建主表时一同创建索引表。
        }
        //client.createTable(request);
        LOG.info("create table[{}] end.",beanDefinition.getTableName());	
	}
	
	/**
	 * 判断表是否已经存在
	 * @param tableName
	 * @return
	 */
	@Deprecated
	private boolean existTable(String tableName) {
//		if(StringUtils.isBlank(tableName)) {
//			throw new NullPointerException("tableName is null");
//		}
//		ListTableResponse response = client.listTable();
//	    System.out.println("表的列表如下：");
//	    for (String tN : response.getTableNames()) {
//	        if (tableName.equalsIgnoreCase(tN)) {
//	        	return true;
//	        }
//	    }
	    return false;
	}

	@Override
	public long saveChat(ChatRecord record) {
		Map<String,String> beanMap = BeanUtil.transBean2Map(record);
		//去除主键
		beanMap.remove("platform_groupId");
		ChatMessage message = new ChatMessage(beanMap);
		return messageOperation.save(record.getPlatform_groupId(), message);
	}

	@Override
	public long saveWhiteBoard(WhiteBoardRecord record) {
		Map<String,String> beanMap = BeanUtil.transBean2Map(record);
		//去除主键
		beanMap.remove("platform_groupId");
		WhiteBoardMessage message = new WhiteBoardMessage(beanMap);
		return messageOperation.save(record.getPlatform_groupId(), message);
	}

	@Override
	public List<Map<String,String>> getHistoryMessage(String platform, String groupId, Long seqId) {
		List<IMessage> messages = messageOperation.getHistoryMessage(platform + ":" + groupId, seqId);
		List<Map<String,String>> list = new ArrayList<>();
		for (IMessage message : messages) {
			list.add(message.getAttributes());
		}
		return list;
	}
	
}
